package com.zhaosc.rx.example;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.schedulers.Schedulers;

public class ColdObservalExample {

	/**
	 * 是只有观察者订阅了，才开始执行发射数据流的代码 。 
	 * 井且 Cold Observabl巳和 Observer 只能是一对一的关系，独享数据。
	 * just、 creat、 range、 fromXXX 等操作符都能生成 Cold Observable
	 */
	@Test
	public void testColdObservable() {
		Observable<Long> observable = Observable.create(new ObservableOnSubscribe<Long>() {
			@Override
			public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
				Observable.interval(10L, TimeUnit.MILLISECONDS, Schedulers.computation()).take(Integer.MAX_VALUE)
						.subscribe(e::onNext);
			}
		}).observeOn(Schedulers.newThread());

		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer1:" + t);
			}
		});


		SleepUtils.sleep(1000L);
		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer2:" + t);
			}
		});

		SleepUtils.sleep(2000L);

	}
	
	/**
	 * Hot Observable 如何转换成 Cold Observable
	 */
	@Test
	public void testHot2ColdObservable() {
		
		ConnectableObservable<Long>  connectableObservable = Observable.create(new ObservableOnSubscribe<Long>() {
			@Override
			public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
				Observable.interval(10L, TimeUnit.MILLISECONDS, Schedulers.computation()).take(Integer.MAX_VALUE)
						.subscribe(e::onNext);
			}
		}).observeOn(Schedulers.newThread()).publish();

		connectableObservable.connect();
		connectableObservable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer1:" + t);
			}
		});
		SleepUtils.sleep(1000L);
		
		connectableObservable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer2:" + t);
			}
		});
		Observable<Long> observable=connectableObservable.refCount();
		
		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer3:" + t);
			}
		});
		

		SleepUtils.sleep(1000L);
		observable.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("consumer4:" + t);
			}
		});


		SleepUtils.sleep(1000L);
	}
}
